用 DSPy GEPA 优化多跳检索问答

多跳检索增强生成，就是一个问题不能靠一次检索回答，必须先查出中间事实，再用这个中间事实继续查
例子：问“David Gregory 继承的城堡有几层”，第一跳查出城堡是 Kinnairdy Castle，第二跳再查 Kinnairdy Castle 有几层

DSPy

DSPy 是把语言模型应用写成可编译程序的框架，开发者用签名声明输入输出，用模块组织模型调用和工具调用，用评价函数定义好坏，再把程序交给优化器改提示词、示例或其他可调部分

手写提示词靠人盯着坏例子反复改词，DSPy 把坏例子、执行轨迹、评价分数放进优化循环，GEPA 是其中一个优化器，它特别依赖自然语言反馈，适合把“哪里错了”转成“下一版指令怎么写”

项目地址：https://github.com/stanfordnlp/dspy
文档地址：https://dspy.ai/

问题

普通检索增强生成通常是一次检索加一次回答：用户问题进入检索器，检索器返回若干文段，语言模型基于文段生成答案，多跳检索问答多了一层状态更新，系统必须先把原问题变成第一跳查询，拿到证据后再生成下一跳查询，再把多轮证据合成答案，常见失败有三类：第一跳查询太宽，第二跳没有利用上一轮证据，回答模块把上下文里没有的事实补出来

GEPA 的入口正是模块提示词，多跳检索问答有天然的模块边界：generate_query 负责根据问题和笔记生成下一跳查询，retrieve 负责拿证据，answer 负责只基于笔记回答，GEPA 不训练模型权重，它把失败执行里的轨迹和评价反馈交给反思模型，让反思模型改某个语言模块的指令

系统形式化定义

上图这段是 GEPA 的系统记法，读法不复杂，Φ 是整套程序，在本文里就是 MultiHopRAG，C 是控制流，决定先调用查询模块、再调用检索器、最后调用回答模块，M_i 是一个语言模块，比如 generate_query 或 answer

每个语言模块写成 M_i = (π_i, θ_i, X_i, Y_i)，π_i 是提示词，θ_i 是模型权重，X_i 和 Y_i 是输入输出格式，GEPA 只改 π_i，不改 θ_i，所以它和 GRPO、权重微调的边界很清楚，权重保持冻结，优化预算用来编译提示词程序

基线程序

先写一个最小多跳检索问答，只保留必要结构，让失败暴露出来，这里的原则很简单：把“生成查询、检索、回答”拆开，GEPA 才能知道该改哪个语言模块的指令

python

class ToyRetriever:
    def __init__(self):
        self.index = {
            "david gregory castle inherited": [
                "David Gregory inherited Kinnairdy Castle through the Gregory family",
            ],
            "kinnairdy castle storeys": [
                "Kinnairdy Castle is a tower house with five storeys",
            ],
        }

    def __call__(self, query):
        q = normalize(query)
        for key, passages in self.index.items():
            if key in q or all(term in q for term in key.split()):
                return passages
        return []


class MultiHopRAG(dspy.Module):
    def __init__(self, retriever, hops=2):
        self.hops = hops
        self.retriever = retriever
        self.generate_query = dspy.ChainOfThought("question, notes -> query")
        self.answer = dspy.ChainOfThought("question, notes -> answer")

    def forward(self, question):
        notes = []
        queries = []

        for _ in range(self.hops):
            query = self.generate_query(question=question, notes=notes).query
            queries.append(query)
            notes.extend(self.retriever(query))

        out = self.answer(question=question, notes=notes)
        return dspy.Prediction(answer=out.answer, context=notes, queries=queries)

这段程序有两个可优化预测器：generate_query 和 answer，检索器是普通工具，不属于 GEPA 要改的语言模块，返回值里保留 queries 和 context，目的是给评价函数做归责，让它判断失败发生在哪一步

一个失败样本可以写成这样：

text

问题:
How many storeys are in the castle David Gregory inherited?

标准答案:
five

查询:
1. David Gregory castle inherited
2. David Gregory castle storeys

取回的证据:
David Gregory was a Scottish mathematician ...
Kinnairdy Castle is associated with the Gregory family ...

模型回答:
three

这条失败主要来自检索链路，第一跳查询找到了人物和城堡关系，但第二跳没有把中间实体 Kinnairdy Castle 明确带入查询，检索器没拿到“Kinnairdy Castle has five storeys”的证据，回答模块只能在证据不足的上下文上猜，给这种样本一个 score=0 没有学习价值，GEPA 需要的是“第二跳查询丢失了中间实体”这种反馈

先看这组对照：

text

坏查询: ['David Gregory castle inherited', 'David Gregory castle storeys']
查询反馈: 检索失败，后一跳查询没有显式携带上一跳发现的中间实体
程序反馈: 最终答案错误，主要原因是检索证据缺失，应优先修正查询生成模块
好查询: ['David Gregory castle inherited', 'Kinnairdy Castle storeys']
得分: 1.0

评价函数

GEPA 的评价函数最好返回分数加文字反馈，这个文字反馈就是反思模型改提示词时读到的批改意见，函数签名里的几个参数要读懂：

参数	中文含义	在多跳检索问答里的作用
`gold`	标注样本	含问题和标准答案
`pred`	程序输出	含模型答案、查询列表、检索上下文
`trace`	整条执行轨迹	可以看到每个模块怎么被调用
`pred_name`	当前要归责的模块名	例如 `generate_query` 或 `answer`
`pred_trace`	当前模块的局部轨迹	用来判断这个模块自己的输入输出是否合理

主要代码如下：

python

def normalize(text):
    return re.sub(r"\s+", " ", text.strip().lower())


def answer_supported(answer, context):
    if not answer or not context:
        return False
    needle = normalize(answer)
    return any(needle in normalize(passage) for passage in context)


def multihop_metric(gold, pred, trace=None, pred_name=None, pred_trace=None):
    exact = normalize(pred.answer or "") == normalize(gold.answer)
    grounded = answer_supported(pred.answer, pred.context)
    has_gold_evidence = any(
        normalize(gold.answer) in normalize(passage)
        for passage in (pred.context or [])
    )

    if exact and grounded:
        return dspy.Prediction(score=1.0, feedback="答案正确，且答案能从上下文推出")

    if pred_name == "generate_query":
        return dspy.Prediction(
            score=0.0,
            feedback=(
                "检索失败，后一跳查询没有显式携带上一跳发现的中间实体，"
                "下一版指令应要求查询使用笔记中的实体和关系继续检索，"
                "例如把 David Gregory inherited castle 转成 Kinnairdy Castle storeys"
            ),
        )

    if pred_name == "answer" and not grounded:
        return dspy.Prediction(
            score=0.0,
            feedback=(
                "回答失败，答案没有被上下文支持，"
                "下一版指令应要求只使用上下文中出现的事实，"
                "证据不足时返回无法判断"
            ),
        )

    if not has_gold_evidence:
        return dspy.Prediction(
            score=0.0,
            feedback="最终答案错误，主要原因是检索证据缺失，应优先修正查询生成模块",
        )

    return dspy.Prediction(
        score=0.0,
        feedback="最终答案错误，上下文中已有证据，但回答模块没有正确抽取",
    )

这段评价函数要写清“归责”，同样是错答，如果上下文里没有标准答案证据，反馈应指向查询模块，如果上下文里已有证据但答案仍错，反馈才应指向回答模块，pred_name 和 pred_trace 就是 GEPA 给模块级反馈预留的接口

编译

DSPy 里的 GEPA 编译入口如下，task_lm 负责执行学生程序，reflection_lm 负责读轨迹和反馈后改指令，两者可以是同一个模型，也可以分开配，反思模型需要足够上下文窗口读完整失败轨迹

python

dspy.configure(lm=task_lm)

gepa = dspy.GEPA(
    metric=multihop_metric,
    max_metric_calls=6,
    reflection_lm=reflection_lm,
    reflection_minibatch_size=1,
    candidate_selection_strategy="pareto",
    component_selector="round_robin",
    use_merge=False,
    add_format_failure_as_feedback=True,
    track_stats=True,
    log_dir="runs/gepa-multihop-rag",
)

student = MultiHopRAG(retriever)
compiled = gepa.compile(student, trainset=trainset, valset=valset)

一次最小编译的输出如下：

text

answer: five
queries: ['David Gregory inherited castle number of storeys', 'Kinnairdy Castle number of storeys']
score: 1.0
metric_calls: 6
candidates: 2

先读输出，不急着庆祝分数，第二跳查询已经转向中间实体 Kinnairdy Castle，这说明 GEPA 的改动落在查询生成模块，回答提示词没有承担主要变化，玩具数据只有一个样本，反思出的指令容易带上样本专名，技术上跑通不等于泛化成立，在真实 HotPotQA 或 HoVer 风格数据上，应提高 max_metric_calls，并让反馈讲“中间实体”和“缺失关系”，不要讲“Kinnairdy Castle”这种专名

普通提示词

单提示词任务也能放进 GEPA 框架，系统 Φ 只有一个语言模块 M_1，候选池保存多版提示词，反思变异每轮只改这一处，分类、抽取、改写、路由、审核，都可以这样建模

判断 GEPA 有没有用，先看评价函数能不能批改样本，“我感觉回答不好”没有学习信号，“标准标签是投诉，模型输出咨询，因为文本里出现退款和超时”才有信号，反思模型读到这种反馈，才知道下一版提示词该提高退款、超时、扣费等触发词权重

普通工单分类：

python

class TicketRouter(dspy.Module):
    def __init__(self):
        self.route = dspy.Predict("ticket -> label")

    def forward(self, ticket):
        return self.route(ticket=ticket)


def route_metric(gold, pred, trace=None, pred_name=None, pred_trace=None):
    if pred.label == gold.label:
        return dspy.Prediction(score=1.0, feedback="分类正确")

    return dspy.Prediction(
        score=0.0,
        feedback=(
            f"分类错误，标准标签是 {gold.label}，模型输出 {pred.label}，"
            "下一版指令要优先识别退款、超时、无法登录、账单扣费这些显式触发词，"
            "不要只根据语气判断标签"
        ),
    )

如果一个提示词一直改不好，先做三件事：把失败样本分组，检查评价函数能否复现你的判断，把反馈写到模块级，这三件事做不到，GEPA 只会把预算花在反复改写措辞上，问题来自知识缺口、工具缺口、检索缺口时，也应先补系统能力，再谈提示词优化

场景	是否该跑 GEPA	判断
A 类样本好了，B 类样本又坏	高	帕累托候选池能保留不同分片上的局部优解
输出格式经常错	中高	格式失败可以变成明确反馈
人工提示词改到第十版还在绕圈	中高	失败轨迹能逼迫改动落到具体规则
没有标注样本，只有主观感觉	低	评价函数不稳定
任务缺知识、缺工具、缺检索	低	提示词无法补齐证据来源

图 2，多跳问答第二跳查询提示词

多跳问答第二跳查询提示词

图 2 直接给了第二跳查询提示词的前后对比，上面的小框是初始提示词，只说“给定 question 和 summary_1，生成 query”，策略几乎为空，下面的大框是 GEPA 优化后的第二跳查询指令，中文拆开就是四件事：读原问题，读第一跳摘要，找第一跳没有覆盖但回答还需要的缺口，生成指向缺口实体的第二跳查询，放到 David Gregory 例子里，第一跳摘要告诉你城堡是 Kinnairdy Castle，第二跳就不该继续搜 David Gregory castle，而该搜 Kinnairdy Castle storeys

这里要看规则密度，不看提示词长度，有效变化是把错误压成规则：不要复述原问题，不要重复已经检索过的事实，要把第一跳摘要里暗示的新实体变成第二跳查询目标

图 4，主算法和帕累托候选选择

主算法和帕累托候选选择

图 4 左边是主算法，右边是候选选择算法，论文符号和 DSPy 代码的对应关系如下：

论文符号	中文解释	在本文代码里的对应
`Φ`	当前完整系统	`MultiHopRAG(retriever)`
`D_train`	训练样本集合	`trainset`
`D_feedback`	用来产生反馈的小样本池	GEPA 抽取小批量样本做反思
`D_pareto`	用来维护候选池排名的验证样本	`valset`
`µ`	评价分数	`multihop_metric(...).score`
`µ_f`	文字反馈函数	`multihop_metric(...).feedback`
`B`	执行预算	`max_metric_calls`
`P`	候选程序池	GEPA 保存的不同提示词版本
`S`	得分矩阵	每个候选在每个验证样本上的分数
`π_j`	第 j 个模块的提示词	`generate_query` 或 `answer` 的指令

把算法 1 翻成工程语言，就是这几步：先用原始程序在验证样本上打分，然后从候选池里选一个程序，选它的一个语言模块，在小批量样本上跑出轨迹、分数、反馈，把这些材料交给反思模型生成新版指令，只替换这个模块的提示词，再跑同一小批量看是否变好，如果变好，就把新程序放回候选池，并在验证样本上记录它的分数

差别就在这里：普通搜索只知道某段提示词得分高不高，GEPA 在每轮变异前会拿到“程序怎么错、哪个模块错、评价函数怎么批改”的轨迹，对多跳检索问答来说，系统可以把“第二跳查询丢失 Kinnairdy Castle”写进查询模块指令，避免把问题泛化成“整条流水线要更准确”

图 3，候选池、帕累托前沿、反思变异和系统合并

候选池、帕累托前沿、反思变异和系统合并

图 3 的英文标签逐一翻译如下：

图中标签	中文读法	工程含义
Candidate Pool	候选程序池	每个候选都是一套完整提示词
Scores Matrix	得分矩阵	行是任务，列是候选，格子是分数
Best candidate per task	每个任务上的最好候选	不看平均分，先看谁解决了某一类题
Filtered Pool	过滤后的候选池	去掉各项都被别的候选压住的版本
Pareto Frontier	帕累托前沿	至少在某些任务上领先的候选集合
Reflective Prompt Mutation	反思式提示词变异	读失败轨迹后改一个模块指令
System Aware Merge	系统感知合并	把两个候选在不同模块上的优点合起来
Performance improved	小批量分数变好	变好才进入完整验证

Scores Matrix 可以想成一张编译日志表：

text

候选 | 父候选 | 改动模块       | 任务1 | 任务2 | 任务3 | 备注
0    | 无     | 原始程序       | 0     | 1     | 0     | 原始提示词
1    | 0      | generate_query | 1     | 1     | 0     | 查询携带中间实体
2    | 0      | answer         | 0     | 1     | 1     | 回答严格依赖上下文
3    | 1,2    | 合并           | 1     | 1     | 1     | 合并查询和回答规则

帕累托选择的意思是：候选 1 平均分可能不高，但它解决了“中间实体传递”这一类题，候选 2 也许排不到全局第一，但它解决了“答案必须有上下文支撑”这一类题，如果只保留平均分最高的候选，这两条局部策略可能被过早丢掉，GEPA 保留这些局部赢家，后续才有机会通过合并得到候选 3

读结果

编译后不要只看最终分数，先看 detailed_results：

python

result = compiled.detailed_results
print(result.total_metric_calls)
print(result.val_aggregate_scores[-5:])
print(result.parents[-5:])
print(result.discovery_eval_counts[-5:])

这些字段的中文解释如下：total_metric_calls 是评价函数调用次数，也就是预算用了多少，val_aggregate_scores 是验证集汇总分数，能看候选整体是否上涨，parents 是候选的父节点，能看提示词是否沿着某条路径演化，discovery_eval_counts 是发现每个候选花掉的评价次数，能看样本效率，per_val_instance_best_candidates 能看不同验证样本是否由不同候选领先，如果所有样本都由同一个候选领先，帕累托机制的收益有限，如果不同候选覆盖不同样本，说明任务里确实存在多种局部策略

分数之后，必须读编译后的提示词，一次有效的查询模块变异，会明确要求把笔记里的具体实体放进查询，并要求查询同时包含答案目标，抽掉样本专名后，可迁移的部分是下面三条：

text

根据笔记里最具体的实体生成下一跳查询
如果笔记已经识别出中间实体，就直接查询这个实体和缺失关系
当更具体的实体已经出现时，不要重复原来的宽泛查询

这类变化才说明 GEPA 学到了可迁移规则，如果提示词只是变成“请更准确并仔细使用上下文”，评价反馈仍然太弱，如果提示词直接写死 David Gregory 和 Kinnairdy Castle，说明样本太少或反馈太贴样本，下一步要扩充 D_feedback，并把评价文案改成规则级语言

图 5，隐私任务里的提示词变异轨迹

隐私任务里的提示词变异轨迹

图 5 来自 PUPA 隐私任务，用来展示 GEPA 的“演化”形态，圆点是候选程序，圆点里的数字是候选编号，括号里的数是分数，红色箭头是从原始候选走向最好候选的路径，旁边的文字记录每次提示词新增的规则，具有诊断价值

按中文读，这条红线大概是：原始指令只说“保护隐私”，候选 2 加入“识别并泛化个人信息”，候选 4 加入“结构化输出和领域规则”，候选 5 加入“解释为什么这样改写”，候选 11 加入“严格步骤协议和零泄露容忍”，迁移到多跳检索问答，理想轨迹也应如此：第一轮学会携带中间实体，第二轮学会避免重复宽泛查询，第三轮学会答案只来自上下文，第四轮学会证据不足时拒答

对照实验

验证 GEPA，不能只展示一条成功样本，最小对照实验要把四种方案放在同一数据、同一预算口径下比较：

方案	调什么	观察什么
零样本基线	不优化	原始分数、失败类型
BootstrapFewShot	示例	是否由样例格式带来提升
MIPROv2	指令和示例	分数和提示词长度是否一起上涨
GEPA	指令和反馈	分数、评价次数、提示词长度、验证集和测试集差距

如果 GEPA 分数上涨，但提示词变得很长，线上成本可能不划算，如果 GEPA 分数不如 MIPROv2，但提示词词元明显更少，要看你的推理成本约束，如果 GEPA 在验证集上涨、测试集不涨，通常是反馈把样本专名写进了指令，或者 D_pareto 太小

执行与成本

GEPA 和 GRPO 执行次数曲线

图 11 的横轴 Number of Rollouts，中文就是完整执行次数：程序跑一遍问题、产生推理和工具调用、再被评价函数打分，算一次，纵轴是测试集分数，蓝线是 GEPA，橙线是 GRPO，这条曲线讲的是样本效率：当评价函数能提供可读诊断时，GEPA 可以用更少执行次数找到有效提示词更新，前提必须守住，执行轨迹和评价反馈里要有可学习信息

最终分数和提示词词元数的关系

图 17 的横轴是汇总提示词词元数，可以粗略当作推理成本代理，纵轴是汇总分数，论文给出的结论是，GEPA 的提示词通常比 MIPROv2 短很多，因为 MIPROv2 常靠少量示例提供行为模式，GEPA 更倾向把失败里抽出的规则写进指令，工程判断不能只看验证分数，还要看提示词词元、延迟、缓存命中率和线上调用成本

排错表

现象	原因	改法
GEPA 只写通用建议	反馈只有分数或空泛文本	在评价函数里写明失败模块、失败条件、下一版规则
查询模块没有变化	程序没有暴露 `generate_query` 预测器	拆模块，检查 `student.named_predictors()`
回答模块乱改	评价函数把检索失败归责给回答模块	用 `has_gold_evidence` 区分检索失败和回答失败
验证集涨、测试集不涨	反馈过拟合样本专名	把专名改成规则，例如写“中间实体”，避免写“Kinnairdy Castle”
提示词变长太多	反思模型把失败逐条塞进指令	在反馈中要求抽象规则，不要求记住样本
GEPA 不如 MIPROv2	任务更依赖示例	先用 MIPROv2 找示例，再用 GEPA 改指令

结论

一句话概括 GEPA：它把程序的一次次失败执行，转写成可归责的自然语言反馈，再用这些反馈演化各个模块的提示词，并用帕累托候选池保留不同题型上的局部优解

放到多跳检索问答里，GEPA 学到的是具体程序规则：第二跳查询要携带上一跳发现的中间实体，回答必须被上下文支撑，证据不足时不要猜测，它比手写提示词更像优化器，因为它从失败轨迹里抽规则，减少凭感觉润色文本

边界也很清楚：GEPA 的上限不由算法名字决定，而由评价函数决定，只给分数，它只是提示词搜索，给出可归责反馈，它才像一个会批改作业的教授，把“错了”改成“哪里错、为什么错、下一版指令该怎么写”

来源

GEPA 论文：https://arxiv.org/abs/2507.19457
DSPy GEPA 文档：https://dspy.ai/api/optimizers/GEPA/overview/
DSPy GEPA Advanced：https://dspy.ai/api/optimizers/GEPA/GEPA_Advanced/

DSPy ​

问题 ​

基线程序 ​

评价函数 ​

编译 ​

普通提示词 ​

读结果 ​

对照实验 ​

执行与成本 ​

排错表 ​

结论 ​

来源 ​

DSPy

问题

基线程序

评价函数

编译

普通提示词

读结果

对照实验

执行与成本

排错表

结论

来源